package com.atguigu.day08;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

//实时对账系统
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<OrderEvent> appStream = env.addSource(new SourceFunction<OrderEvent>() {
            @Override
            public void run(SourceContext<OrderEvent> ctx) throws Exception {
                OrderEvent o1 = new OrderEvent("order1", "app", 1000L);
                OrderEvent o2 = new OrderEvent("order2", "app", 2000L);
                OrderEvent o3 = new OrderEvent("order3", "app", 3000L);
                ctx.collectWithTimestamp(o1, o1.timestamp);
                ctx.collectWithTimestamp(o2, o2.timestamp);
                ctx.collectWithTimestamp(o3, o3.timestamp);
            }

            @Override
            public void cancel() {

            }
        });

        DataStreamSource<OrderEvent> wxStream = env.addSource(new SourceFunction<OrderEvent>() {
            @Override
            public void run(SourceContext<OrderEvent> ctx) throws Exception {
                OrderEvent o1 = new OrderEvent("order1", "wx", 4000L);
                OrderEvent o2 = new OrderEvent("order3", "wx", 5000L);
                OrderEvent o3 = new OrderEvent("order2", "wx", 9000L);
                ctx.collectWithTimestamp(o1, o1.timestamp);
                ctx.collectWithTimestamp(o2, o2.timestamp);
                ctx.emitWatermark(new Watermark(8000L));
                ctx.collectWithTimestamp(o3, o3.timestamp);
            }

            @Override
            public void cancel() {

            }
        });
        appStream
                .keyBy(x -> x.orderId)
                .connect(
                        wxStream.keyBy(x -> x.orderId)
                )
                .process(new MatchProcess())
                .print();


        env.execute();
    }

    public static class MatchProcess extends CoProcessFunction<OrderEvent, OrderEvent, String> {
        private ValueState<OrderEvent> appState;
        private ValueState<OrderEvent> wxState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            appState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("app", Types.POJO(OrderEvent.class)));
            wxState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("wx", Types.POJO(OrderEvent.class)));
        }

        /**
         * app处理流
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
            //当微信流没来时 保存app流信息到状态变量 并注册定时器
            System.out.println(ctx.timerService().currentWatermark());
            if (wxState.value() == null) {
                appState.update(value);
                ctx.timerService().registerEventTimeTimer(value.timestamp + 5000L);
            }else {
                //若微信流先来   则对账成功 清除wx状态变量
                out.collect(value.orderId+"对账成功");
                wxState.clear();
            }
        }

        /**
         * wx处理流
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement2(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
            //当app流没来时 保存wx流信息到状态变量 并注册定时器
            System.out.println(ctx.timerService().currentWatermark());
            if (appState.value() == null) {
                wxState.update(value);
                ctx.timerService().registerEventTimeTimer(value.timestamp + 5000L);
            }else {
                //若app流先来则app状态变量有值   则对账成功 清除app状态变量
                out.collect(value.orderId+"对账成功");
                appState.clear();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            //5秒之后  app依旧没来数据  清空wx状态变量数据
            if (wxState.value()!=null){
                out.collect(wxState.value().orderId+"对账失败 app数据未到达 时间："+timestamp);
                wxState.clear();
            }
            //5秒之后  wx依旧没来数据  清空app状态变量数据
            if (appState.value()!=null){
                out.collect(appState.value().orderId+"对账失败 wx数据未到达 时间："+timestamp);
                appState.clear();
            }
        }
    }


    public static class OrderEvent {
        public String orderId;
        public String type;
        public Long timestamp;

        public OrderEvent() {
        }

        public OrderEvent(String orderId, String type, Long timestamp) {
            this.orderId = orderId;
            this.type = type;
            this.timestamp = timestamp;
        }
    }
}
